package com.iot.service.mqtt.handler;

import com.iot.service.mqtt.bridge.TlBridgeService;
import com.iot.service.mqtt.config.Constant;
import com.iot.service.mqtt.model.fix.TlMqttFixedHead;
import com.iot.service.mqtt.model.payload.TlMqttPublishPayload;
import com.iot.service.mqtt.model.request.TlMqttPubRecReq;
import com.iot.service.mqtt.model.request.TlMqttPublishReq;
import com.iot.service.mqtt.model.response.TlMqttPubAck;
import com.iot.service.mqtt.model.variable.TlMqttPublishVariableHead;
import com.iot.service.mqtt.service.retain.IRetainService;
import com.iot.service.mqtt.utils.ChannelManager;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.AttributeKey;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
@RequiredArgsConstructor
public class TlMqttPublishHandler extends SimpleChannelInboundHandler<TlMqttPublishReq> {

    private static final Logger logger = LoggerFactory.getLogger(TlMqttPublishHandler.class);

    private final IRetainService retainMsgService;

    private final TlForwardMessageService forwardMessageService;

    private final TlBridgeService bridgeService;

    private final ChannelManager channelManager;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttPublishReq req) throws Exception {

        logger.debug("in【PUBLISH】 handler");
        Channel channel = ctx.channel();
        Object clientId = channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get();
        if (clientId == null) {
            channel.close();
            return;
        }


        TlMqttFixedHead fixedHead = req.getFixedHead();
        TlMqttPublishVariableHead variableHead = req.getVariableHead();
        TlMqttPublishPayload payload = req.getPayload();

        boolean retain = fixedHead.isRetain();
        MqttQoS messageQos = fixedHead.getQos();
        String topic = variableHead.getTopic();
        Long messageId = variableHead.getMessageId();
        String content = payload.getContent().toString();

        logger.debug("【{}】 send topic 【{}】 qos【{}】 messageId 【{}】", clientId, topic, messageQos, messageId);
        /*如果是保留消息 存储*/
        if (retain) {
            storeRetain(topic, content, messageQos, messageId).subscribe();
        }
        //数据转发到其他服务
        //bridgeService.bridge(clientId.toString(), messageQos.value(), req);
        switch (messageQos) {
            case AT_LEAST_ONCE -> sendAck(messageId, clientId.toString());
            case EXACTLY_ONCE -> {
                //这里需要保存消息 key是messageId，value是req，在收到rel消息后 需要将这个消息转发到其他订阅的客户端 在rel只能收到messageId，没有其他的信息
                channel.attr(AttributeKey.valueOf(Constant.PUB_MSG)).set(req);
                //发送rec消息给发送者
                sendRec(messageId, clientId.toString());
                return;
            }
            default -> logger.error("error qos level");
        }
        //转发给其他的订阅的客户端
        forwardMessageService.forwardMessage(topic,  messageQos, content);
    }

    /**
     * @description: 在新订阅的时候发送
     * @author: hszhou
     * @datetime: 2025-04-29 18:29:08
     * @param:
     * @param: topic 主题
     * @param: content 内容
     * @param: qos
     * @param: messageId
     * @return: Mono<Boolean>
     **/
    private Mono<Boolean> storeRetain(String topic, String content, MqttQoS qos, Long messageId) {

        return Mono.defer(()->{
            if (StringUtils.isEmpty(content)) {
                return retainMsgService.clear(topic);
            }else{
                TlMqttPublishReq pubMsg = TlMqttPublishReq.build(topic, qos, true, content, messageId);
                return retainMsgService.save(topic, pubMsg);
            }
        });
    }


    /**
     * 构建ack消息发送返回
     *
     * @param messageId 消息ID
     * @param clientId  客户端ID
     */
    private void sendAck(Long messageId, String clientId) {
        TlMqttPubAck res = TlMqttPubAck.build(messageId);
        logger.debug("send ack messageId【{}】", res);
        channelManager.writeAndFlush(clientId,res);
    }

    /**
     * @description: 发送rec消息给客户端
     * @author: hszhou
     * @datetime: 2025-05-08 16:21:42
     * @param:
     * @param: messageId
     * @param: clientId
     * @return: void
     **/
    private void sendRec( Long messageId, String clientId) {
        TlMqttPubRecReq res = TlMqttPubRecReq.build(messageId);
        channelManager.writeAndFlush(clientId,res);
    }
}
